package org.example;

import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

public class RedisHashSink implements RedisMapper<Order> {

    /**
     * 设置redis数据类型
     */
    @Override
    public RedisCommandDescription getCommandDescription() {
        /**
         * 第1个参数是Hash数据类型, 第二个参数是外面的key
         *
         * @redisCommand: Hash数据类型
         * @additionalKey: 哈希和排序集数据类型时使用(RedisDataType.HASH组或RedisDataType.SORTED_SET), 其他类型忽略additionalKey
         */
        return new RedisCommandDescription(RedisCommand.HSET, "redis");
    }

    /**
     * 设置Key
     */
    @Override
    public String getKeyFromData(Order oder) {
        return oder.getId() + "";
    }

    /**
     * 设置value
     */
    @Override
    public String getValueFromData(Order oder) {
        // 从数据中获取Value: Hash的value
        return JSON.toJSONString(oder);
    }
}
